package mr.stocks.question2;

import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Iterator;
import java.util.List;

import mr.stocks.avro.model.StocksKeyAvro;
import mr.stocks.avro.model.StocksReportAvro;
import mr.stocks.avro.model.StocksValuesAvro;
import mr.stocks.model.StocksInputWritable;

import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapred.AvroValue;
import org.apache.avro.mapreduce.AvroMultipleOutputs;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import au.com.bytecode.opencsv.CSVReader;

public class StocksAvroUpdateMapper
		extends
		Mapper<AvroKey<StocksReportAvro>, NullWritable, AvroKey<StocksReportAvro>, NullWritable> {
	private static Logger LOG = LoggerFactory.getLogger(StocksAvroUpdateMapper.class);

	private Calendar cal = Calendar.getInstance();
	private static final String SYMBOL_DESCRIPTION_FILE = "stocks_updates.csv";
	private static final double delta = 0.00001;

	private List<StocksValuesAvro> updateList = new ArrayList<StocksValuesAvro>();
	private StocksValuesAvro updateEntry;
	private String yearStr_stockUpdate = new String();
	private String monthStr_stockUpdate = new String();
	private String dayStr_stockUpdate = new String();
	
	private String yearStr = new String();
	private String monthStr = new String();
	private String dayStr = new String();

	@Override
	protected void setup(Context context) throws IOException,
			InterruptedException {

		FileReader reader = null;
		CSVReader csvReader = null;
		try {
			reader = new FileReader(SYMBOL_DESCRIPTION_FILE);
			csvReader = new CSVReader(reader);
			List<String[]> entries = csvReader.readAll();
			Iterator<String[]> i = entries.iterator();

			while (i.hasNext()) {
				updateEntry = new StocksValuesAvro();
				String[] tokens = i.next();
				updateEntry.setStock(tokens[0]);
				updateEntry.setTimestamp(Long.parseLong(tokens[1]));
				updateEntry.setPrice(Double.parseDouble(tokens[2]));
				updateEntry.setStockVolume(Long.parseLong(tokens[3]));
				updateList.add(updateEntry);
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			IOUtils.closeQuietly(csvReader);
			IOUtils.closeQuietly(reader);
		}
	}
	
	@Override
	protected void map(AvroKey<StocksReportAvro> key, NullWritable inValue,
			Context context) throws IOException, InterruptedException {

		boolean found = false;
		StocksKeyAvro date = key.datum().getYearMonthDay();
		this.yearStr = date.getYear().toString();
		this.monthStr = date.getMonth().toString();
		this.dayStr = date.getDay().toString();
		
		for (StocksValuesAvro stock : updateList) {

			cal.setTimeInMillis(stock.getTimestamp());
			yearStr_stockUpdate = Integer.toString(cal.get(Calendar.YEAR));
			monthStr_stockUpdate = Integer.toString(cal.get(Calendar.MONTH) + 1);
			dayStr_stockUpdate = Integer.toString(cal.get(Calendar.DAY_OF_MONTH));

			if (yearStr_stockUpdate.equals(yearStr)
					&& monthStr_stockUpdate.equals(monthStr)
					&& dayStr_stockUpdate.equals(dayStr)) {
				List<StocksValuesAvro> stockList = key.datum().getStocks();
				for (StocksValuesAvro value : stockList) {
					if (value.getStock().toString()
							.equals(stock.getStock().toString())
							&& Math.abs(value.getPrice() - stock.getPrice()) < delta) {
						LOG.info("Found Match");
						found = true;
						value.setStockVolume(stock.getStockVolume());
					}
				}
			}
		}

		if(found){
			context.write(key, NullWritable.get());
		}
	}
}
